package com.reger.canal.core;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.protobuf.InvalidProtocolBufferException;
import com.reger.canal.client.MessageProcess;
import com.reger.canal.client.StringUnderlineCamelUtils;

public class DefaultMessageProcess implements MessageProcess {
	
	private static final Logger log = LoggerFactory.getLogger(DefaultMessageProcess.class);

	protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
	protected String destination;
	
	@Autowired RabbitTemplate rabbitTemplate;

	public void process(Message message) {
		
		List<Entry> entrys = message.getEntries();
		try {
			for (Entry entry : entrys) {
				this.printEntry(entry);
			}
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
	}

	protected void printEntry(Entry entry) throws InvalidProtocolBufferException {
		Header header = entry.getHeader();
		log.debug(String.format("事件类型{} , 变更数据的执行时间 {} ,event的长度{} , binlog文件名 {}, binlog偏移量{} ,数据库名{} , 变更数据的编码{} ,mysql服务器编号{},协议版本号 {} "
				,entry.getEntryType()
				, header.getExecuteTime()
				, header.getEventLength()
				, header.getLogfileName()
				, header.getLogfileOffset()
				, header.getSchemaName()
				, header.getServerenCode()
				, header.getServerId()
				, header.getVersion()
				));
		
		switch (entry.getEntryType()) {
		case ROWDATA:  // 二进制数据事件
			this.rowChage(entry);
		break;
		case HEARTBEAT: // 心跳数据事件
			this.heartbeat(entry);
		break;
		case TRANSACTIONBEGIN: //事务开始事件
			this.transactionBegin(entry);
		break;
		case TRANSACTIONEND:   //事务结束事件
			this.transactionEnd(entry);
		break;

		default:
			break;
		}
	}
	
	private void heartbeat(Entry entry) throws InvalidProtocolBufferException {
//		HeartBeat heartBeat= HeartBeat.parseFrom(entry.getStoreValue());
	}

	private void transactionBegin(Entry entry) throws InvalidProtocolBufferException {
//		TransactionBegin begin= TransactionBegin.parseFrom(entry.getStoreValue());
//		System.err.println(begin.getThreadId());
	}
	
	private void transactionEnd(Entry entry) throws InvalidProtocolBufferException {
//		TransactionEnd end= TransactionEnd.parseFrom(entry.getStoreValue());
//		System.err.println(end.getTransactionId());
	}
	
	private void rowChage(Entry entry) throws InvalidProtocolBufferException {
		String dataBaseName=entry.getHeader().getSchemaName();
		String tableName=entry.getHeader().getTableName();
		RowChange rowChage = RowChange.parseFrom(entry.getStoreValue());
//		rowChage.getTableId();
//		rowChage.getDdlSchemaName();
		EventType eventType = rowChage.getEventType();
		String queueName = String.format("REDER.DATEBUS.QUEUE.%s.%s.%s", dataBaseName, tableName, eventType).toUpperCase();
//		if (eventType == EventType.QUERY || rowChage.getIsDdl()) {
//			System.err.println(" sql ----> " + rowChage.getSql());
//			return;
//		}

		for (RowData rowData : rowChage.getRowDatasList()) {
			final Map<String, Map<String, Object>> map = this.toMaps(rowData, eventType);
			log.debug(" ------>> [{}]  >>>> {}" ,queueName,new Object(){@Override public String toString() { return DefaultMessageProcess.this.toJsonString(map); }} );
			rabbitTemplate.convertAndSend(queueName, null, map);
		}
	}

	private Map<String, Map<String, Object>> toMaps(RowData rowData, EventType eventType) {
		Map<String, Map<String, Object>> map = new HashMap<String, Map<String, Object>>();
		switch (eventType) {
		case INSERT: // 插入数据
			map.put("after", this.toMap(rowData.getAfterColumnsList()));
			break;
		case UPDATE: // 更新数据
			map.put("before", this.toMap(rowData.getBeforeColumnsList()));
			map.put("after", this.toMap(rowData.getAfterColumnsList()));
			break;
		case DELETE: // 删除数据
			map.put("before", this.toMap(rowData.getBeforeColumnsList()));
			break;
		case CREATE: // 创建表

			break;
		case ALTER: // 更新表
			map.put("before", this.toMap(rowData.getBeforeColumnsList()));
			map.put("after", this.toMap(rowData.getAfterColumnsList()));
			break;
		case ERASE: // 删除表
			map.put("before", this.toMap(rowData.getBeforeColumnsList()));
			break;
		case QUERY: // 查询数据

			break;
		case TRUNCATE: // 截/断表
			map.put("after", this.toMap(rowData.getAfterColumnsList()));
			map.put("before", this.toMap(rowData.getBeforeColumnsList()));
			break;
		case RENAME: // 重命名表

			break;
		case CINDEX: // 创建索引

			break;
		case DINDEX: // 删除索引

			break;

		default:
			break;
		}
		return map;
	}

	private String toJsonString(Object value) {
		try {
			return new ObjectMapper().writeValueAsString(value);
		} catch (JsonProcessingException e) {
			e.printStackTrace();
			return null;
		}
	}

	protected Map<String, Object> toMap(List<Column> columns) {

		Map<String, Object> value = new HashMap<String, Object>();
		for (Column column : columns) {
			if (column.getIsNull()) {
				continue;
			}
			String name = StringUnderlineCamelUtils.toCamel2(column.getName());
			String val = column.getValue();
			value.put(name, val);
		}
		return value;
	}
}
